package com.xxwu.messagequeue.mqserver.core;

import com.xxwu.messagequeue.common.Consumer;
import com.xxwu.messagequeue.common.ConsumerEnv;
import com.xxwu.messagequeue.common.MqException;
import com.xxwu.messagequeue.mqserver.VirtualHost;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;

/**
 * 用来执行消费消息的核心逻辑
 */
@Slf4j

public class ConsumerManager {
    // 将两个类关联起来方便方法的相互调用
    private VirtualHost virtualHost;

    // 线程池，用来执行消费者提供的回调函数
    private ExecutorService workThread = Executors.newFixedThreadPool(4);
    private BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<String>();
    // 扫描线程
   private Thread scannerThread = null;

   public int getBlockingQueueSize() {
       return blockingQueue.size();
   }

    // 使用构造方法将这两个类关联起来
    public ConsumerManager(VirtualHost virtualHost){
        this.virtualHost = virtualHost;

        scannerThread = new Thread(()->{
            while(true){
                try{
                    // 从阻塞队列中获取令牌
                    String queueName = blockingQueue.take();
                    MSGQueue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);
                    if(queue == null){
                        throw new MqException("该队列不存在！queueName=" + queueName);
                    }
                    // 去这个队列里消费一个消息
                    consumeMessage(queue);
                } catch (MqException | InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        // 将扫描线程设置为后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    // 向阻塞队列中添加消息,当向队列发送一条消息时就调用该方法
    public void addMessage(String queueName){
        blockingQueue.add(queueName);
    }

    // 将消费者添加到队列
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        MSGQueue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);
        if(queue == null){
            throw new MqException("该队列不存在 queueName= "+ queueName);
        }
        // 创建消费者
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        // 将消费者添加到指定队列
        synchronized (queue){
            queue.addConsumerEnv(consumerEnv);
            // 如果队列中已经有消息了就要立刻消费掉
            int messageNums = virtualHost.getMemoryDataCenter().getMessageNums(queueName);
            for(int i = 0; i < messageNums; i++){
                // 调用一次这个方法就消费一条消息
                consumeMessage(queue);
            }
        }
    }

    // 消费消息
    private void consumeMessage(MSGQueue queue) {
        // 取出一个消费者
        ConsumerEnv consumerEnv = queue.getConsumerEnv();
        //这一步拿出来就没有Tag 估计还在上面
        if(consumerEnv == null){
            // 该队列没有消费者订阅，先不消费，等有了再消费
            return;
        }
        // 取出一条消息
        Message message = virtualHost.getMemoryDataCenter().getMessageQueue(queue.getName());
        // 若消息不存在，先不消费
        if(message == null){
            return;
        }
        workThread.submit(()->{
            try{
                // 将消息添加到 等待应答的消息队列中
                virtualHost.getMemoryDataCenter().insertMessageQueueWaitAck(queue.getName(), message);
                // 消费消息,执行消费者的回调函数
                log.info("consumerEnv={}", consumerEnv);

                consumerEnv.getConsumer().handleDelivery(consumerEnv.getConsumerTag(), message.getBasicProperties(), message.getBody());
                // 判断消费者采用的消息应答方式
                // 消费者采用自动应答
                if(consumerEnv.isAutoAck()){
                    // 将消息是否持久化
                    if(message.getDeliveryMode() == 2){
                        // 将消息从硬盘中删除
                        virtualHost.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    // 将消息从内存中删除
                    virtualHost.getMemoryDataCenter().deleteMessageQueueWaitAck(queue.getName(), message);
                    // 这里不用考虑还要从队列中删除消息的操作，当消息从队列中取出的时候就已经不在队列中了
                    // 而且这里如果消费者的回调方法执行失败，初始化的时候还会从硬盘中将数据拷贝会内存
                    // 不至于造成消息丢失
                    virtualHost.getMemoryDataCenter().deleteMessage(message.getMessageId());
                    log.info("消息被消费成功！");
                }
            }catch (Exception e){
                e.printStackTrace();
                log.error("消费消息失败");
            }
        });
    }
}
